package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/4
 * @Desc: 学大数据，上多易教育
 *
 *   topic1:用户行为数据
 *      {"uid":4,"event_id":"page_load","properties":{"url":"/b","ref":"/x"},"action_time":1704719574005}
 *   topic2: 用户注册信息
 *      {"uid":4,"gender":"male","birthday":"2000-03-06","register_time":1704719879001}
 *
 *   将用户行为和用户注册信息，按照相同用户每 5分钟窗口来关联
 *
 **/
public class _13_Exercise_JOIN {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(1000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().set("table.exec.source.idle-timeout","1s");
        tenv.getConfig().set("table.exec.state.ttl","1 hour");


        // {"uid":1,"event_id":"face_recg","action_time":10000}
        tenv.executeSql(
                " create table action_log_kfk(                                    "+
                        "   uid int,                                            "+
                        " 	event_id string,                                    "+
                        " 	properties map<string,string>,                      "+
                        " 	action_time bigint,                                 "+
                        " 	pt as proctime(),                                   "+
                        " 	rt as to_timestamp_ltz(action_time,3),              "+
                        " 	watermark for rt as rt                              "+
                        "  ) with (                                             "+
                        "    'connector' = 'kafka',                             "+
                        "    'topic' = 'action-log',                            "+
                        "    'properties.bootstrap.servers' = 'doitedu:9092',   "+
                        "    'properties.group.id' = 'g003',                    "+
                        "    'scan.startup.mode' = 'latest-offset',             "+
                        "    'value.format' = 'json',                           "+
                        "    'value.fields-include' = 'EXCEPT_KEY'              "+
                        " )                                                     "
        );

        // {"uid":1,"face_status":1,"ts":11000}
        tenv.executeSql(
                " create table register_info_kfk(                                    "+
                        "   uid int,                                            "+
                        " 	gender string,                                      "+
                        " 	birthday date,                                      "+
                        " 	register_time bigint,                               "+
                        " 	pt as proctime(),                                   "+
                        " 	rt as to_timestamp_ltz(register_time,3),            "+
                        " 	watermark for rt as rt                              "+
                        "  ) with (                                             "+
                        "    'connector' = 'kafka',                             "+
                        "    'topic' = 'register-info',                         "+
                        "    'properties.bootstrap.servers' = 'doitedu:9092',   "+
                        "    'properties.group.id' = 'g003',                    "+
                        "    'scan.startup.mode' = 'latest-offset',             "+
                        "    'value.format' = 'json',                           "+
                        "    'value.fields-include' = 'EXCEPT_KEY'              "+
                        " )                                                     "
        );

        tenv.executeSql(
                " WITH tmp1 AS (                                                                  "+
                        "     SELECT                                                                      "+
                        "         uid,                                                                    "+
                        "         event_id,                                                               "+
                        "         properties,                                                               "+
                        "         action_time,                                                            "+
                        "         window_start,                                                           "+
                        "         window_end                                                              "+
                        "     FROM TABLE(                                                                 "+
                        "         TUMBLE(TABLE action_log_kfk, DESCRIPTOR(rt),INTERVAL '10' SECOND)         "+
                        "     )                                                                           "+
                        " ),                                                                              "+
                        " tmp2 AS (                                                                       "+
                        "     SELECT                                                                      "+
                        "         uid,                                                                    "+
                        "         gender,                                                                 "+
                        "         birthday,                                                               "+
                        "         register_time,                                                          "+
                        "         window_start,                                                           "+
                        "         window_end                                                              "+
                        "     FROM TABLE(                                                                 "+
                        "         TUMBLE(TABLE register_info_kfk, DESCRIPTOR(rt),INTERVAL '10' SECOND)    "+
                        "     )                                                                           "+
                        " )                                                                               "+
                        "                                                                                 "+
                        " SELECT                                                                          "+
                        "     tmp1.uid,                                                                   "+
                        "     tmp1.event_id,                                                              "+
                        "     tmp1.properties,                                                             "+
                        "     tmp1.action_time,                                                           "+
                        "     tmp1.window_start,                                                          "+
                        "     tmp1.window_end,                                                            "+
                        "     tmp2.gender,                                                                "+
                        "     tmp2.birthday,                                                              "+
                        "     tmp2.register_time                                                          "+
                        " FROM tmp1 JOIN tmp2                                                             "+
                        " ON tmp1.window_start = tmp2.window_start                                        "+
                        " AND tmp1.window_end = tmp2.window_end                                           "+
                        " AND tmp1.uid = tmp2.uid                                                         "
        ).print();



    }







}
